package cn.boat.router.broker;

import cn.boat.router.MetaZookeeper;
import cn.boat.router.broker.lock.DistributedLock;
import cn.boat.router.broker.lock.impl.SimpleDistributedLockImpl;
import cn.boat.router.models.*;
import cn.boat.router.utils.JSONUtils;
import cn.boat.router.utils.ZkClientUtils;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

public class BrokerZookeeper {

    private Logger logger = LoggerFactory.getLogger(BrokerZookeeper.class);

    // zookeeper客户端
    private ZkClient zkClient;

    // zookeeper辅助类
    private MetaZookeeper metaZookeeper;

    // 分布式锁
    private DistributedLock distributedLock;

    private ConfigCenter configCenter;

    public BrokerZookeeper(ZkClient zkClient, MetaZookeeper metaZookeeper) {
        this.zkClient = zkClient;
        this.metaZookeeper = metaZookeeper;
        this.distributedLock = new SimpleDistributedLockImpl(zkClient, "/locker");
        configCenterInit();
    }

    /**
     * 配置中心初始化
     */
    private void configCenterInit() {
        if (!zkClient.exists(metaZookeeper.configCenterPath)) {
            ConfigCenter configCenter = new ConfigCenter(
                    "com.mysql.jdbc.Driver",
                    "jdbc:mysql://localhost:3306/order?useUnicode=true&amp;characterEncoding=UTF-8",
                    "root",
                    "123465"
            );
            try {
                ZkClientUtils.createPersistentPath(zkClient, metaZookeeper.configCenterPath, JSONUtils.serializeObject(configCenter));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else {
            updateConfigCenter();
        }
        zkClient.subscribeDataChanges(metaZookeeper.configCenterPath, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                updateConfigCenter();
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                logger.error("请注意!配置中心所有数据被删除!");
            }
        });
    }

    /**
     * 注册broker节点（实际）
     * @param brokerId ID
     */
    public boolean registerBroker(int brokerId, String brokerHost, int brokerPort) throws Exception {

        String newBrokerPath = metaZookeeper.brokerIdsPath+"/"+brokerId;
        // 判断节点是否存在
        if (ZkClientUtils.pathExists(zkClient, newBrokerPath)){
            return false;
        }
        String serializeData = JSONUtils.serializeObject(new BrokerModel(brokerHost, brokerPort));

        // 加分布式锁
        distributedLock.acquire();
        ZkClientUtils.createEphemeralPath(zkClient, newBrokerPath, serializeData);
        distributedLock.release();

        return true;
    }
    /**
     * 注册持久broker节点（供测试使用）
     * @param brokerId ID
     */
    public boolean registerPersistentBroker(int brokerId, String brokerHost, int brokerPort) throws Exception {

        String newBrokerPath = metaZookeeper.brokerIdsPath+"/"+brokerId;
        // 判断节点是否存在
        if (ZkClientUtils.pathExists(zkClient, newBrokerPath)){
            return false;
        }
        ZkClientUtils.makeSurePersistentPathExists(zkClient, newBrokerPath);
        String serializeData = JSONUtils.serializeObject(new BrokerModel(brokerHost, brokerPort));
        ZkClientUtils.updatePersistentPath(zkClient, newBrokerPath, serializeData);
        return true;
    }


    /**
     * 根据brokerId获取其所管理的主题、分区
     * @param brokerId brokerId
     * @return
     * @throws Exception
     */
    public List<TopicAndPartition> getBrokerMapById(Integer brokerId) throws Exception {
        List<Topic> allTopicNameToPartitionsMap = getAllTopicNameToPartitionsMap();
        List<TopicAndPartition> result = new ArrayList<>();
        for (Topic topic : allTopicNameToPartitionsMap) {
            List<Partition> partitionList = topic.getPartitionList();
            for (Partition partition : partitionList) {
                if (partition.getIsr().contains(brokerId)) {
                    TopicAndPartition topicAndPartition = new TopicAndPartition(topic.getTopicName(), partition, partition.getLeader()==brokerId);
                    result.add(topicAndPartition);
                }
            }
        }
        return result;
    }


    /**
     * 获取所有最新的brokerMetaData
     * @return BrokerMetaData
     */
    public BrokerMetaData getAllBrokerMetaData() throws Exception {
        List<Broker> allBrokerMetaData = getAllBrokerData();
        List<Topic> allTopicNameToPartitionsMap = getAllTopicNameToPartitionsMap();
        return new BrokerMetaData(allBrokerMetaData, allTopicNameToPartitionsMap);
    }

    /**
     * 获取所有的broker
     * @return
     */
    public List<Broker> getAllBrokerData() throws Exception {
        List<Broker> brokerResult = new LinkedList<>();
        List<String> children = zkClient.getChildren(metaZookeeper.brokerIdsPath);
        for(String brokerId : children){
            String brokerData = zkClient.readData(metaZookeeper.brokerIdsPath + "/" + brokerId);
            BrokerModel brokerModel = (BrokerModel)JSONUtils.deserializeObject(brokerData, BrokerModel.class);
            Broker newBroker = new Broker(Integer.parseInt(brokerId), brokerModel.getBrokerHost(), brokerModel.getBrokerPort());
            brokerResult.add(newBroker);
        }
        return brokerResult;
    }



    /**
     * 张
     * 获取所有主题、分区及其对应关系
     * @return
     * @throws Exception
     */
    public List<Topic> getAllTopicNameToPartitionsMap() throws Exception {
        List<Topic> topicList = new LinkedList<>();
        List<String> topicNames = new ArrayList<>();
        if (zkClient.exists(metaZookeeper.brokerTopicsPath)) {
            topicNames = zkClient.getChildren(metaZookeeper.brokerTopicsPath);
        }
        for (String topicName : topicNames) {
            List<String> partitionNames = zkClient.getChildren(metaZookeeper.brokerTopicsPath + "/" + topicName+"/partitions");
            List<Partition> partitionList = new LinkedList<>();
            for (String partitionId : partitionNames) {
                String partitionData = zkClient.readData(metaZookeeper.brokerTopicsPath + "/" + topicName + "/partitions/" + partitionId + "/state");
                PartitionModel partitionModel = (PartitionModel)JSONUtils.deserializeObject(partitionData, PartitionModel.class);
                Partition newPartition = new Partition(Integer.parseInt(partitionId), partitionModel.getIsr().size(),
                        partitionModel.getLeader(), partitionModel.getIsr());
                partitionList.add(newPartition);
            }
            Topic newTopic = new Topic(topicName, partitionNames.size(), partitionList);
            topicList.add(newTopic);
        }
        return topicList;
    }

    /**
     * 张
     * 返回给定topic的路由信息
     * @param topicName topic名字
     */
    public Topic getTopicByName(String topicName) throws Exception {
        List<Topic> allTopicNameToPartitionsMap = getAllTopicNameToPartitionsMap();
        for (Topic topic: allTopicNameToPartitionsMap) {
            if (topic.getTopicName().equals(topicName)) {
                return topic;
            }
        }
        return null;
    }

    /**
     * 返回给定brokerId的broker信息
     * @param brokerId brokerId
     * @return  Broker
     * @throws Exception
     */
    public Broker getBrokerById(Integer brokerId) throws Exception {
        List<Broker> allBrokerData = getAllBrokerData();
        for (Broker broker: allBrokerData) {
            if (broker.getBrokerId() == brokerId) {
                return broker;
            }
        }
        return null;
    }

    /**
     * 获取全局唯一ID
     * @return ID
     */
    public String getUniqueId() throws Exception {
        distributedLock.acquire();
        String ephemeralSequential = zkClient.createEphemeralSequential(metaZookeeper.uniquePath, "aa");
        distributedLock.release();
        return ephemeralSequential.substring(metaZookeeper.uniquePath.length()+1);
    }


    /**
     * 更新配置中心数据
     */
    public void updateConfigCenter(){
        try {
            String data = zkClient.readData(metaZookeeper.configCenterPath);
            configCenter = (ConfigCenter) JSONUtils.deserializeObject(data, ConfigCenter.class);
            logger.info("自动获取到配置中心数据："+configCenter);
        }catch (Exception e) {

            e.printStackTrace();
        }
    }

    /**
     * 安全关闭
     */
    public void close() throws InterruptedException {
        zkClient.unsubscribeAll();
        Thread.sleep(100);
        zkClient.close();
        logger.info("zookeeper 连接安全关闭");
    }


    public static void main(String[] args) throws Exception {
        // 该对象放到客户端配置类中
        ZkClientUtils.ZKConfig zkConfig = new ZkClientUtils.ZKConfig();


        ZkClient zkClient =
                new ZkClient(zkConfig.zkConnect, zkConfig.zkSessionTimeoutMs,
                        zkConfig.zkConnectionTimeoutMs, new ZkClientUtils.StringSerializer());

        BrokerZookeeper bz = new BrokerZookeeper(zkClient, new MetaZookeeper(zkClient, "/boatMQ"));

        ZkClientUtils.updatePersistentPath(zkClient, "/get", "bbb");
//        System.out.println(bz.getBrokerMapById(0));
//
//
//        System.out.println(bz.getAllBrokerMetaData());

        bz.registerPersistentBroker(0, "192.168.128.1", 5147);
        bz.registerPersistentBroker(1, "192.168.12.15", 2159);
        bz.registerPersistentBroker(2, "192.168.78.55", 6666);

//        System.out.println(bz.getAllBrokers());

//        if (!bz.registerBroker(3,"192.168.122.2",8456)) {
//            System.out.println("创建失败！");
//        }
        Thread.sleep(100000);
    }
}
